#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "ipcapi.h"

// url示例:
// 	"tcp://127.0.0.1:9901"
// 	"tcp://127.0.0.1:9902"
// 	"ipc:///tmp/reqrep"  /* 这种写法要求路径/tmp能够创建reqrep文件 */
// 	"ipc:///tmp/pubsub"
static char REQREP_SERVER[256] = {0};
static char PUBSUB_SERVER[256] = {0};

static void* reqrepServer1 = NULL;
static void* pubsubServer2 = NULL;

static void* reqrepClient1 = NULL;
static void* reqrepClient2 = NULL;
static void* pubsubClient3 = NULL;
static void* pubsubClient4 = NULL;

static int32_t gTestExit = 0;

#if defined(WIN32) || defined(WIN64) || defined(WINCE) || defined(_MSC_VER)
/* ----- windows begin ----- */
#include <windows.h>
#include <process.h>
#define PLAT_THREAD_TYPE void
#define PLAT_THREAD_RETURN
#define SLEEP(s) Sleep(s * 1000)
static void CreateThread(void* obj, void (*callback)(void*))
{
	_beginthread(callback, 0, obj);
}
/* ----- windows end ----- */
#else
/* ----- linux begin ----- */
#include <pthread.h>
#include <unistd.h>
#define PLAT_THREAD_TYPE void*
#define PLAT_THREAD_RETURN NULL
#define SLEEP(s) sleep(s)
static void CreateThread(void* obj, void* (*callback)(void*))
{
	pthread_t th;
	pthread_attr_t attr;
	// attr init
	pthread_attr_init(&attr);
	//禁用线程同步, 线程运行结束后自动释放
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
	//抛出线程
	pthread_create(&th, &attr, callback, (void*)obj);
	// attr destroy
	pthread_attr_destroy(&attr);
}
/* ----- linux end ----- */
#endif

static void MsgCallback(void* handle, void* priv, char* buff, int32_t buffLen)
{
	if (priv == &reqrepServer1)
	{
		fprintf(stdout, "reqrepServer1 recv len/%d, buff/%s \r\n", buffLen, buff);
		IpcApiSend(reqrepServer1, buff, buffLen, 0);
	}
	else if (priv == &pubsubServer2)
	{
		fprintf(stdout, "pubsubServer2 recv len/%d, buff/%s \r\n", buffLen, buff);
	}
	else if (priv == &reqrepClient1)
	{
		fprintf(stdout, "reqrepClient1 recv len/%d, buff/%s \r\n", buffLen, buff);
	}
	else if (priv == &reqrepClient2)
	{
		fprintf(stdout, "reqrepClient2 recv len/%d, buff/%s \r\n", buffLen, buff);
	}
	else if (priv == &pubsubClient3)
	{
		fprintf(stdout, "pubsubClient3 recv len/%d, buff/%s \r\n", buffLen, buff);
	}
	else if (priv == &pubsubClient4)
	{
		fprintf(stdout, "pubsubClient4 recv len/%d, buff/%s \r\n", buffLen, buff);
	}
}

PLAT_THREAD_TYPE Server(void* argv)
{
	int32_t ret;
	int32_t count = 0;
	char buff[128] = {0};
	fprintf(stdout, " ---------- Server begin ---------- \r\n");

	reqrepServer1 = IpcApiServer(REQREP_SERVER, IPCAPI_TYPE_REQREP, &reqrepServer1, &MsgCallback);
	if (!reqrepServer1) {
		fprintf(stderr, "reqrepServer1 IpcApiServer %s failed \r\n", REQREP_SERVER);
	}

	pubsubServer2 = IpcApiServer(PUBSUB_SERVER, IPCAPI_TYPE_PUBSUB, &pubsubServer2, &MsgCallback);
	if (!pubsubServer2) {
		fprintf(stderr, "pubsubServer2 IpcApiServer %s failed \r\n", PUBSUB_SERVER);
	}

	while (!gTestExit)
	{
		SLEEP(1);
		
		snprintf(buff, sizeof(buff), "ab|%04d", count);
		ret = IpcApiSend(pubsubServer2, buff, strlen(buff), 0);
		if (ret < 1) {
			fprintf(stderr, "pubsubServer2 send \"%s\" failed \r\n", buff);
		}

		snprintf(buff, sizeof(buff), "12|%04d", count);
		ret = IpcApiSend(pubsubServer2, buff, strlen(buff), 0);
		if (ret < 1) {
			fprintf(stderr, "pubsubServer2 send \"%s\" failed \r\n", buff);
		}

		count += 1;
	}

	IpcApiRelease(reqrepServer1);
	IpcApiRelease(pubsubServer2);
	
	fprintf(stdout, " ---------- Server exit ---------- \r\n");
    return PLAT_THREAD_RETURN;
}

PLAT_THREAD_TYPE Client(void* argv)
{
	int32_t ret;
	char buff[1024] = { 0 };
	fprintf(stdout, " ---------- Client begin ---------- \r\n");

	reqrepClient1 = IpcApiClient(REQREP_SERVER, IPCAPI_TYPE_REQREP, NULL, NULL);
	if (!reqrepClient1) {
		fprintf(stderr, "reqrepClient1 IpcApiClient %s failed \r\n", REQREP_SERVER);
	}

	reqrepClient2 = IpcApiClient(REQREP_SERVER, IPCAPI_TYPE_REQREP, NULL, NULL);
	if (!reqrepClient2) {
		fprintf(stderr, "reqrepClient2 IpcApiClient %s failed \r\n", REQREP_SERVER);
	}

	pubsubClient3 = IpcApiClient(PUBSUB_SERVER, IPCAPI_TYPE_PUBSUB, &pubsubClient3, &MsgCallback);
	if (!pubsubClient3) {
		fprintf(stderr, "pubsubClient3 IpcApiClient %s failed \r\n", PUBSUB_SERVER);
	}
	IpcApiSetTopic(pubsubClient3, "ab");

	pubsubClient4 = IpcApiClient(PUBSUB_SERVER, IPCAPI_TYPE_PUBSUB, &pubsubClient4, &MsgCallback);
	if (!pubsubClient4) {
		fprintf(stderr, "pubsubClient4 IpcApiClient %s failed \r\n", PUBSUB_SERVER);
	}
	IpcApiSetTopic(pubsubClient4, "12");

	while (!gTestExit)
	{
		ret = IpcApiSend(reqrepClient1, (char*)"1234", 4, 0);
		if (ret < 1) {
			fprintf(stderr, "reqrepClient1 send \"1234\" failed \r\n");
		}

		ret = IpcApiSend(reqrepClient2, (char*)"5678", 4, 0);
		if (ret < 1) {
			fprintf(stderr, "reqrepClient2 send \"5678\" failed \r\n");
		}

		ret = IpcApiRecv(reqrepClient2, buff, sizeof(buff), 500);
		if (ret < 1) {
			fprintf(stderr, "reqrepClient2 recv failed \r\n");
		}
		else {
			fprintf(stdout, "reqrepClient2 recv len/%d, buff/%s \r\n", ret, buff);
		}

		ret = IpcApiRecv(reqrepClient1, buff, sizeof(buff), 500);
		if (ret < 1) {
			fprintf(stderr, "reqrepClient1 recv failed \r\n");
		}
		else {
			fprintf(stdout, "reqrepClient1 recv len/%d, buff/%s \r\n", ret, buff);
		}

		SLEEP(1);
	}

	IpcApiRelease(reqrepClient1);
	IpcApiRelease(reqrepClient2);
	IpcApiRelease(pubsubClient3);
	IpcApiRelease(pubsubClient4);

	fprintf(stdout, " ---------- Client exit ---------- \r\n");
    return PLAT_THREAD_RETURN;
}

int main()
{
	fprintf(stdout, " ---------- Create url ---------- \r\n");

	IpcApiCreateUrl(IPCAPI_TYPE_REQREP, REQREP_SERVER, sizeof(REQREP_SERVER), "reqrep-test");
	IpcApiCreateUrl(IPCAPI_TYPE_PUBSUB, PUBSUB_SERVER, sizeof(PUBSUB_SERVER), "pubsub-test");

	printf("REQREP_SERVER: %s\r\n", REQREP_SERVER);
	printf("PUBSUB_SERVER: %s\r\n", PUBSUB_SERVER);

	SLEEP(1);

	gTestExit = 0;
	CreateThread(NULL, &Server);
	SLEEP(1);
	CreateThread(NULL, &Client);
	for (int i = 0; i < 5; i++) {
		SLEEP(1);
	}
	gTestExit = 1;

	SLEEP(2);

	fprintf(stdout, " ---------- Enter to exit ---------- \r\n");
	getchar();
	return 0;
}
